package com.lry.basic.metrics;

import com.google.common.eventbus.Subscribe;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

public class MetricsSubscriber {

    private final Registry registry;

    private final Map<GlobalStatus, Consumer<GlobalTransactionEvent>> consumers;

    public MetricsSubscriber(Registry registry) {
        this.registry = registry;
        consumers = new HashMap<>();

        consumers.put(GlobalStatus.Begin,this::processGlobalStatusBegin);
        consumers.put(GlobalStatus.Committed, this::processGlobalStatusCommitted);
        consumers.put(GlobalStatus.Rollbacked, this::processGlobalStatusRollbacked);
        consumers.put(GlobalStatus.CommitFailed, this::processGlobalStatusCommitFailed);
        consumers.put(GlobalStatus.RollbackFailed, this::processGlobalStatusRollbackFailed);
        consumers.put(GlobalStatus.TimeoutRollbacked, this::processGlobalStatusTimeoutRollbacked);
        consumers.put(GlobalStatus.TimeoutRollbackFailed, this::processGlobalStatusTimeoutRollbackFailed);
    }

    //counter(计数，例如计算当前处于active的全局事务个数),summary（统计max,count,total,tps）,timer(max,avg,count)

    //开启全局事务，增加当前处于active的全局事务个数
    private void processGlobalStatusBegin(GlobalTransactionEvent event) {
        registry.getCounter(MeterIdConstants.COUNTER_ACTIVE).increase(1);
    }

    //提交全局事务
    //减少当前处于active的全局事务个数
    //增加当前处于committed的全局事务个数
    private void processGlobalStatusCommitted(GlobalTransactionEvent event) {
        registry.getCounter(MeterIdConstants.COUNTER_ACTIVE).decrease(1);
        registry.getCounter(MeterIdConstants.COUNTER_COMMITTED).increase(1);
        registry.getSummary(MeterIdConstants.SUMMARY_COMMITTED).increase(1);
        registry.getTimer(MeterIdConstants.TIMER_COMMITTED).record(event.getEndTime() - event.getBeginTime(),
                TimeUnit.MILLISECONDS);
    }

    //回滚全局事务
    //减少当前处于active的全局事务个数
    //增加当前处于rollbacked的全局事务个数
    private void processGlobalStatusRollbacked(GlobalTransactionEvent event) {
        registry.getCounter(MeterIdConstants.COUNTER_ACTIVE).decrease(1);
        registry.getCounter(MeterIdConstants.COUNTER_ROLLBACKED).increase(1);
        registry.getSummary(MeterIdConstants.SUMMARY_ROLLBACKED).increase(1);
        registry.getTimer(MeterIdConstants.TIMER_ROLLBACK).record(event.getEndTime() - event.getBeginTime(),
                TimeUnit.MILLISECONDS);
    }

    //全局事务提交失败
    //减少当前处于active的全局事务个数
    private void processGlobalStatusCommitFailed(GlobalTransactionEvent event) {
        registry.getCounter(MeterIdConstants.COUNTER_ACTIVE).decrease(1);
    }

    //全局事务回滚失败
    //减少当前处于active的全局事务个数
    private void processGlobalStatusRollbackFailed(GlobalTransactionEvent event) {
        registry.getCounter(MeterIdConstants.COUNTER_ACTIVE).decrease(1);
    }

    private void processGlobalStatusTimeoutRollbacked(GlobalTransactionEvent event) {
        registry.getCounter(MeterIdConstants.COUNTER_ACTIVE).decrease(1);
    }

    private void processGlobalStatusTimeoutRollbackFailed(GlobalTransactionEvent event) {
        registry.getCounter(MeterIdConstants.COUNTER_ACTIVE).decrease(1);
    }

    @Subscribe
    public void recordGlobalTransactionEventForMetrics(GlobalTransactionEvent event){
        if (registry != null && consumers.containsKey(event.getStatus())) {
            consumers.get(event.getStatus()).accept(event);
        }
    }

}
